package my.thesis.spouts;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.util.ArrayList;
import java.util.Map;

import my.thesis.analytics.metrics.LastTimePersistMetric;
import my.thesis.analytics.metrics.ThroughputMetric;
import my.thesis.analytics.metrics.whereAmIMetric;

import org.joda.time.DateTime;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.ReportedFailedException;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class HttpStream extends BaseRichSpout 
{
	private static final long serialVersionUID = 1L;
	private BufferedReader reader;
	private SpoutOutputCollector _collector;
	
	transient whereAmIMetric _whereMetric;
	private static Boolean metricsOn=false;
	
	//private int failedRetries=0;
	private String URL;

	//ME ENDIAMESO BUFFER
	private ArrayList<String> toSend = new ArrayList<String>();
	//int i=1; //using for msgID
		
	public HttpStream (String connectionString)
	{
		this.URL=connectionString;
	}
	
	public void nextTuple() 
	{		
		try 
		{
			//ME BUFFERED READER
			String in;
			in = this.reader.readLine();
			if(in!=null)
			{
					//DEN DOULEVEI!!!
					///System.out.println(in);
					//DateTime timestamp = DateTime.now();
					//long millis = timestamp.getMillis();
					//_collector.emit(new Values(in,millis));
				    
				    //ME ENDIAMESO BUFFER
					toSend.add(in);
			}
			else 
			{
					//System.out.println("End of file reached bb...");
					Thread.sleep(10);
/*					failedRetries++;
					if(failedRetries > 10)
					{
						System.out.println("End of file reached bb...");
						
						//throw new ReportedFailedException("End Of file reached... bb");
						//Error(String message)
						//Constructs a new error with the specified detail message.
						
						Thread.sleep(10);	
					}*/			
			}
			
			for(String send : toSend)
			{								
				//_collector.emit(new Values(send),i); 
				_collector.emit(new Values(send)); 
			}
			
			toSend.clear();
			
			//i += 1;
		}
		
		catch (Exception e)
		{
			System.err.println("Error reading line "+e.getMessage());
			//throw new ReportedFailedException(e);
		}
	}

	public void open(Map map, TopologyContext context, SpoutOutputCollector outputCollector) 
	{
		if(metricsOn)
		{
			_whereMetric=new whereAmIMetric();
			context.registerMetric("spoutMetric", _whereMetric, 20);
			_whereMetric.declare();
		}
		
		this._collector = outputCollector;
		
		try 
		{
			URL url = new URL(URL);
			this.reader = new BufferedReader(new InputStreamReader(url.openStream()));	
		}
		catch (IOException e)
		{
			System.err.println("Error in communication with server ["+e.getMessage()+"]");
		}	
	}

	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
	{
		outputFieldsDeclarer.declare(new Fields("jsonString"));	
	}
}
